package kafka

import (
	"github.com/Shopify/sarama"
	"github.com/gogf/gf/encoding/gjson"
	"github.com/gogf/gf/frame/g"
	"github.com/gogf/gf/util/gconv"
	"errors"
)

// delete records by topic partition offsets
func (c *KafkaClient) DeleteRecords(topic string, partitionOffsets map[int32]int64) error {
	err := c.KAdmin().DeleteRecords(topic, partitionOffsets)
	if err != nil {
		return errors.New("create partitions fail, err:" + err.Error())
	}
	return nil
}

// update consumer offset
func (c *KafkaClient) UpdateConsumerOffset(group, topic string, partition int32, offset int64) error {
	client := c.KClient()
	oc, err := sarama.NewOffsetManagerFromClient(group, client)
	if err != nil {
		return errors.New("new offset mange client fail, err:" + err.Error())
	}
	pom, _ := oc.ManagePartition(topic, partition)
	pom.MarkOffset(offset, "")
	pom.ResetOffset(offset, "")
	return nil
}

// sync consumer groups
func (c *KafkaClient) SyncConsumerGroups(new_group, old_group, topic string) error {
	client := c.KClient()
	oc, err := sarama.NewOffsetManagerFromClient(new_group, client)
	if err != nil {
		return errors.New("new offset mange client fail, err:" + err.Error())
	}
	descs := c.ConsumerGroupsDescribe(old_group, topic)
	for _, v := range descs {
		desc := gconv.Map(v)
		pom, _ := oc.ManagePartition(topic, gconv.Int32(desc["partition"]))
		pom.MarkOffset(gconv.Int64(desc["offset"]), "")
		pom.ResetOffset(gconv.Int64(desc["offset"]), "")
	}
	return nil
}

// consumer group describe
func (c *KafkaClient) ConsumerGroupsDescribe(group, topic string) []map[string]interface{} {
	client := c.KClient()
	parts, _ := client.Partitions(topic)
	result, _ := c.KAdmin().ListConsumerGroupOffsets(group, map[string][]int32{
		topic: parts,
	})

	blocks := gjson.New(gconv.String(result)).GetMap("Blocks." + topic)
	var descList g.List
	for v := range parts {
		end, _ := client.GetOffset(topic, gconv.Int32(v), sarama.OffsetNewest)
		start, _ := client.GetOffset(topic, gconv.Int32(v), sarama.OffsetOldest)
		offset := gjson.New(blocks[gconv.String(v)]).GetInt64("Offset")
		var lag int64
		lag = end - offset
		if offset == -1 {
			lag = end
		}

		offsetMap := g.Map{
			"topic":     topic,
			"group":     group,
			"partition": v,
			"start":     start,
			"end":       end,
			"offset":    offset,
			"lag":       lag,
		}
		descList = append(descList, offsetMap)
	}

	return descList
}
